package com.kc.common.utils;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.kc.common.config.RocketMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;


/**
 * RocketMQ发送消息通用模板工具类
 *
 * @author TP
 * @date 1/12/21 7:03 PM
 */
@Slf4j
@Component
public class RocketMQTemplate {

    @Autowired
    private RocketMqConfig rocketMqConfig;

    @Autowired
    private ProducerBean producer;

    /**
     * 异步发送普通消息（请尽量少使用）
     *
     * @param tag  消息标签（必填）
     * @param data 消息内容（必填）
     */
    public void sendCommonMessageAsync(String tag, Object data) {
        check(tag, data);
        Message message = convertToRocketMessage(rocketMqConfig.getCommonTopic(), tag, null, data, 0);
        log.info(">>> 开始异步发送普通MQ消息，消息内容：{}", message);
        producer.sendAsync(message, new SendCallback() {
            @Override
            public void onSuccess(final SendResult sendResult) {
                assert sendResult != null;
                log.info(">>> 异步发送普通MQ消息成功，消息ID：{}，消息内容：{}", sendResult.getMessageId(), message);
            }

            @Override
            public void onException(final OnExceptionContext context) {
                //TODO 出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
                log.error(">>> 异步发送普通MQ消息失败，要发送的消息内容：{}，异常信息：", message, context.getException());
            }
        });

    }

    /**
     * 同步发送普通消息
     *
     * @param tag  消息标签（必填）
     * @param data 消息内容（必填）
     * @return 发送结果
     */
    public SendResult sendCommonMessage(String tag, Object data) {
        check(tag, data);
        Message message = convertToRocketMessage(rocketMqConfig.getCommonTopic(), tag, null, data, 0);
        log.info(">>> 开始同步发送普通MQ消息，消息内容：{}", message);
        SendResult sendResult = producer.send(message);
        if (sendResult != null) {
            log.info(">>> 同步发送普通MQ消息成功，消息ID：{}，消息内容：{}", sendResult.getMessageId(), message);
        } else {
            log.error(">>> 同步发送普通MQ消息失败，要发送的消息内容：{}", message);
        }
        return sendResult;
    }

    /**
     * 同步发送普通消息
     *
     * @param tag  消息标签（必填）
     * @param data 消息内容（必填）
     * @return 发送结果
     */
    public SendResult sendCommonMessageWithKey(String tag, String key, Object data) {
        check(tag, data);
        if (StringUtils.isEmpty(key)) {
            throw new RuntimeException("message key cannot be empty");
        }
        Message message = convertToRocketMessage(rocketMqConfig.getCommonTopic(), tag, key, data, 0);
        log.info(">>> 开始同步发送普通带Key的MQ消息，消息内容：{}", message);
        SendResult sendResult = producer.send(message);
        if (sendResult != null) {
            log.info(">>> 同步发送普通带Key的Q消息成功，消息ID：{}，消息内容：{}", sendResult.getMessageId(), message);
        } else {
            log.error(">>> 同步发送普通带Key的MQ消息失败，要发送的消息内容：{}", message);
        }
        return sendResult;
    }

    /**
     * 发送延时消息
     *
     * @param tag          消息标签（必填）
     * @param data         消息内容（必填）
     * @param delaySeconds 延时时长（必填，单位秒）
     * @return 发送结果
     */
    public SendResult sendDelayMessage(String tag, Object data, long delaySeconds) {
        check(tag, data);
        if (delaySeconds <= 0) {
            throw new RuntimeException("delay message delaySeconds must > 0");
        }
        Message message = convertToRocketMessage(rocketMqConfig.getDelayTopic(), tag, null, data, delaySeconds);
        log.info(">>> 开步发送延迟MQ消息，消息内容：{}", message);
        SendResult sendResult = producer.send(message);
        if (sendResult != null) {
            log.info(">>> 延时MQ消息发送成功，消息ID：{}，消息内容：{}", sendResult.getMessageId(), message);
        } else {
            log.error(">>> 延时MQ消息发送失败，要发送的消息内容：{}", message);
        }
        return sendResult;
    }

    /**
     * 发送带Key延时消息
     *
     * @param tag          消息标签（必填）
     * @param data         消息内容（必填）
     * @param delaySeconds 延时时长（必填，单位秒）
     * @return 发送结果
     */
    public SendResult sendDelayMessageWithKey(String tag, String key, Object data, long delaySeconds) {
        check(tag, data);
        if (StringUtils.isEmpty(key)) {
            throw new RuntimeException("message key cannot be empty");
        }
        if (delaySeconds <= 0) {
            throw new RuntimeException("delay message delaySeconds must > 0");
        }
        Message message = convertToRocketMessage(rocketMqConfig.getDelayTopic(), tag, key, data, delaySeconds);
        log.info(">>> 开始发送带Key延迟MQ消息，消息内容：{}", message);
        SendResult sendResult = producer.send(message);
        if (sendResult != null) {
            log.info(">>> 带Key延时MQ消息发送成功，消息ID：{}，消息内容：{}", sendResult.getMessageId(), message);
        } else {
            log.error(">>> 带Key延时MQ消息发送失败，要发送的消息内容：{}", message);
        }
        return sendResult;
    }

    private Message convertToRocketMessage(String topic, String tag, String key, Object data, long delaySeconds) {
        byte[] body;
        try {
            if (data instanceof String) {
                body = ((String) data).getBytes(Charset.forName("UTF-8"));
            } else if (data instanceof byte[]) {
                body = (byte[]) data;
            } else {
                String jsonObj = JSON.toJSONString(data);
                body = jsonObj.getBytes(Charset.forName("UTF-8"));
            }
            Message message = new Message(topic, tag, body);
            if (StringUtils.isNotEmpty(key)) {
                message.setKey(key);
            }
            if (delaySeconds > 0) {
                long startDeliverTime = System.currentTimeMillis() + delaySeconds * 1000;
                message.setStartDeliverTime(startDeliverTime);
            }
            return message;

        } catch (Exception e) {
            throw new RuntimeException("convert to RocketMQ message failed. ", e);
        }
    }

    private void check(String tag, Object data) {
        if (StringUtils.isEmpty(tag)) {
            throw new RuntimeException("message tag cannot be empty");
        }
        if (data == null) {
            throw new RuntimeException("message body cannot be empty");
        }
    }
}
